package kafka

import (
	"context"
	"encoding/json"
	"github.com/segmentio/kafka-go"
	"im/models"
	"log"
	"time"
)

var ImConsumer *ImConsumerClient

// Consumer 接口定义了消费者需要实现的方法
type Consumer interface {
	Start(ctx context.Context, handlers map[string]ActionHandler) error
}

// ConsumerClient 结构体表示一个Kafka消费者
type ImConsumerClient struct {
	reader *kafka.Reader
}

// NewConsumeClient 函数用于创建并初始化一个新的ConsumeClient实例
func NewConsumerClient() {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:        []string{"localhost:9092"},
		GroupID:        "im_consumer_group",
		Topic:          "im",
		MinBytes:       10e3, // 10KB
		MaxBytes:       10e6, // 10MB
		CommitInterval: 1 * time.Second,
	})
	ImConsumer = &ImConsumerClient{
		reader: reader,
	}
	go ImConsumer.Start(context.Background(), handlers)
}

// Start 方法启动消费者并开始读取消息，根据actionType调用不同的处理函数
func (c *ImConsumerClient) Start(ctx context.Context, handlers map[string]ActionHandler) error {
	for {
		var kafkaMsg models.CustomKafkaMsg
		msg, err := c.reader.ReadMessage(ctx)
		if err != nil {
			continue
		}
		if err = json.Unmarshal(msg.Value, &kafkaMsg); err != nil {
			continue
		}
		log.Println("ImConsumer.Start 开始消费,msg:", kafkaMsg)
		if err = c.startWorker(ctx, &kafkaMsg, handlers); err != nil {
			log.Println("ImConsumer.Start 消费失败,err:", err)
			continue
		} else {
			// 手动提交offset
			// 注意：通常你会在消息被成功处理后提交offset
			//暂时这样  会出现重复消费，日后修改
			if err := c.reader.CommitMessages(ctx, msg); err != nil {
				log.Printf("Failed to commit offset: %v", err)
			}
		}
	}
}

// startWorker 启动处理特定 actionType 的 goroutine
func (c *ImConsumerClient) startWorker(ctx context.Context, kafkaMsg *models.CustomKafkaMsg, handlers map[string]ActionHandler) error {
	handlerFunc := handlers[kafkaMsg.ActionType]
	return handlerFunc(ctx, kafkaMsg)
}
